package com.whoami.dataplatform.tag_combine.hbase.rw;

import com.google.protobuf.ServiceException;
import com.whoami.dataplatform.tag_combine.hbase.FilterSum;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;

import java.io.IOException;
import java.util.Map;

/**
 * 24 209 & 13199
 * 24 208 & 29973
 * 24 209 & 24 208 & | 43172
 * 24 208 & 209 & 0
 *
 * @author tzp
 * @since 2019/8/14
 */
public class CallFilterSum {
    private static final String TABLE_NAME = "test_tu";
    private static final String CF_DEFAULT = "cf1";

    public static void main(String[] args) throws IOException {
        Configuration config = HBaseConfiguration.create();
        config.addResource(new Path(System.getenv("HBASE_CONF_DIR") + "/conf", "hbase-site.xml"));

        try (Connection connection = ConnectionFactory.createConnection(config)) {
            TableName tableName = TableName.valueOf(TABLE_NAME);
            Table table = connection.getTable(tableName);

            FilterSum.FilterSumRequest request = FilterSum.FilterSumRequest.newBuilder()
                    .setExp("24 ^").build();

            try {
                long start = System.currentTimeMillis();
                Map<byte[], Long> results = table.coprocessorService(
                        FilterSum.FilterSumService.class,
                        null,  /* start key */
                        null,  /* end   key */
                        new Batch.Call<FilterSum.FilterSumService, Long>() {
                            @Override
                            public Long call(FilterSum.FilterSumService aggregate) throws IOException {
                                BlockingRpcCallback<FilterSum.FilterSumResponse> rpcCallback = new
                                        BlockingRpcCallback<>();
                                aggregate.getSum(null, request, rpcCallback);
                                FilterSum.FilterSumResponse response = rpcCallback.get();
                                return response.hasSum() ? response.getSum() : 0L;
                            }
                        });
                for (Long sum : results.values()) {
                    System.out.println("Sum = " + sum);
                }
                System.out.println("time: " + (System.currentTimeMillis() - start)/1000);
            } catch (ServiceException e) {
                e.printStackTrace();
            } catch (Throwable e) {
                e.printStackTrace();
            }
        } catch (Exception e) {

        }
    }
}
